package test;


import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;

import java.util.HashMap;
import java.util.HashSet;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import cn.com.bsfit.frms.obj.AvgNumber;
import cn.com.bsfit.frms.obj.ComparablePair;
import cn.com.bsfit.frms.obj.ConvenientHashMap;
import cn.com.bsfit.frms.obj.CountNumber;
import cn.com.bsfit.frms.obj.DecreaseCountNumber;
import cn.com.bsfit.frms.obj.DistinctedListObject;
import cn.com.bsfit.frms.obj.DurationParsingObject;
import cn.com.bsfit.frms.obj.EventObject;
import cn.com.bsfit.frms.obj.EventRecognitionPattern;
import cn.com.bsfit.frms.obj.IncreaseCountNumber;
import cn.com.bsfit.frms.obj.MaxContinuousCountNumber;
import cn.com.bsfit.frms.obj.MaxDecreaseCountNumber;
import cn.com.bsfit.frms.obj.MaxIncreaseCountNumber;
import cn.com.bsfit.frms.obj.MaxNumber;
import cn.com.bsfit.frms.obj.MaxObject;
import cn.com.bsfit.frms.obj.MemCachedItem;
import cn.com.bsfit.frms.obj.MergeableListObject;
import cn.com.bsfit.frms.obj.MergeableMapObject;
import cn.com.bsfit.frms.obj.MergeableSortedList;
import cn.com.bsfit.frms.obj.MinNumber;
import cn.com.bsfit.frms.obj.MinObject;
import cn.com.bsfit.frms.obj.OccupiedObject;
import cn.com.bsfit.frms.obj.ReplacedObject;
import cn.com.bsfit.frms.obj.Risk;
import cn.com.bsfit.frms.obj.SumNumber;
import cn.com.bsfit.frms.obj.TimedItems;
import cn.com.bsfit.frms.obj.Tuple;
import cn.com.bsfit.frms.obj.VarpNumber;
import cn.com.bsfit.frms.obj.VarsNumber;
import cn.com.bsfit.frms.serial.MemCachedItemKryoSerializer;
import cn.com.bsfit.frms.serial.MemCachedItemUtils;
import cn.com.bsfit.frms.serial.TimedItemsKryoSerializer;

import com.aerospike.client.AerospikeClient;
import com.aerospike.client.AerospikeException;
import com.aerospike.client.Key;
import com.aerospike.client.Record;
import com.aerospike.client.ResultCode;
import com.aerospike.client.ScanCallback;
import com.aerospike.client.policy.ScanPolicy;
import com.aerospike.client.policy.WritePolicy;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.pool.KryoFactory;
import com.esotericsoftware.kryo.pool.KryoPool;

public class AeroSpikeTest {
	private AerospikeClient client = null;
    private KryoPool pool = null;
    private Kryo kryo;
    private Logger log=Logger.getLogger(AeroSpikeTest.class);
    private static int count=0;
    @Before
    public void setUp() throws Exception {
        client = new AerospikeClient("10.100.1.74", 3000);//10.100.1.212
        KryoFactory factory = new KryoFactory() {
            public Kryo create() {
            	Kryo localKryo = new Kryo();
				localKryo.register(MemCachedItem.class, new MemCachedItemKryoSerializer());
			    localKryo.register(TimedItems.class, new TimedItemsKryoSerializer());
			    localKryo.register(AvgNumber.class);
			    localKryo.register(CountNumber.class);
			    localKryo.register(SumNumber.class);
			    localKryo.register(MinNumber.class);
			    localKryo.register(MaxNumber.class);
			    localKryo.register(VarpNumber.class);
			    localKryo.register(VarsNumber.class);
			    localKryo.register(IncreaseCountNumber.class);
			    localKryo.register(DecreaseCountNumber.class);
			    localKryo.register(MaxIncreaseCountNumber.class);
			    localKryo.register(MaxDecreaseCountNumber.class);
			    localKryo.register(MaxContinuousCountNumber.class);
			    localKryo.register(ReplacedObject.class);
			    localKryo.register(OccupiedObject.class);
			    localKryo.register(MaxObject.class);
			    localKryo.register(MinObject.class);
			    localKryo.register(MergeableListObject.class);
			    localKryo.register(MergeableMapObject.class);
			    localKryo.register(DistinctedListObject.class);
			    localKryo.register(MergeableSortedList.class);
			    localKryo.register(EventObject.class);
			    localKryo.register(EventRecognitionPattern.class);
			    localKryo.register(DurationParsingObject.class);
			    localKryo.register(ComparablePair.class);
			    localKryo.register(Tuple.class);
			    localKryo.register(ConvenientHashMap.class);
			    localKryo.register(TreeMap.class);
			    localKryo.register(Risk.class);
			    localKryo.register(Date.class);
			    localKryo.register(HashMap.class);
			    localKryo.register(ConcurrentHashMap.class);
			    localKryo.register(ArrayList.class);
			    localKryo.register(HashSet.class);
				return localKryo;
            }
        };
        pool = new KryoPool.Builder(factory).softReferences().build();
        kryo = pool.borrow();
    }

    @After
    public void tearDown() throws Exception {
    }

    @Test
    public void get() throws ParseException {
    	SimpleDateFormat sdf = new SimpleDateFormat(
				"yyyy-MM-dd HH:mm:ss");
    	System.out.println(new Date().getTime());
    	Key key = new Key("bsfit", "frms", MemCachedItem.getMemCachedKey("tk0001", "支付账号", "ONLINE",true));
		Record rec = null;
		rec = client.get(null, key)
		MemCachedItem item = MemCachedItemUtils.asMemCachedItem(rec, kryo);
    	Object cache = null;
		cache = item.getRaw("用户注册手机归属地");
    	TimedItems ti = (TimedItems)cache;
    	Object obj = ti.getRaw(new Date(), "30d");
    	System.out.println(cache);
    	client.close();
    }
    
    @Test
    public void storeData() {
		Key key = new Key("bsfit", "frms", MemCachedItem.getMemCachedKey("zkk123", "用户号", "PAY.BUY"));
		MemCachedItem memCachedItem = new MemCachedItem("zkk123","用户号","PAY.BUY");
		memCachedItem.put("用户注册手机归属地", "上海");
		client.put(client.writePolicyDefault, key, MemCachedItemUtils.toBins(memCachedItem,kryo));
		
		Key key2 = new Key("bsfit", "frms", MemCachedItem.getMemCachedKey("zkk234", "用户号", "PAY.BUY"));
		MemCachedItem memCachedItem2 = new MemCachedItem("zkk234","用户号","PAY.BUY");
		memCachedItem2.put("用户注册手机归属地", "海南");
		client.put(client.writePolicyDefault, key2, MemCachedItemUtils.toBins(memCachedItem2,kryo));
    	client.close();
	}
    
    @Test
    public void deleteSet(){
    	try {
		   ScanPolicy scanPolicy = new ScanPolicy();
		   client.scanAll(scanPolicy, "bsfit", "etl", new ScanCallback() {

		       public void scanCallback(Key key, Record record) throws AerospikeException {
		           /*
		            * for each Digest returned, delete it using delete()
		            */
		           client.delete(new WritePolicy(), key);
		           count++;
		           /*
		            * after 25,000 records delete, return print the count.
		            */
		           if (count % 1000 == 0){
		               log.info("Deleted "+ count);
		           }
		       }
		   }, new String[] {});
		   System.out.println("Deleted "+ count + " records from set frms");
		   log.info("Deleted "+ count + " records from set frms");
		} catch (AerospikeException e) {
		   int resultCode = e.getResultCode();
		   log.info(ResultCode.getResultString(resultCode));
		   log.debug("Error details: ", e);
		}
    }
}
